package net.bwie.realtime.jtp.dwd.log.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.jtp.common.utils.DateTimeUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * 使用状态state编程，对is_new进行值进行校验和纠偏
 *
 * @Author: FuHe
 * @Date: 2025/5/18
 */

/**
 * 数据采集时，此字段值，依据app在本机缓存的值进行判断赋值的，如果清除缓存或卸载后重新安装的话，
 * 此时字段值不准确，所以需要进行修复。
 * 1). app卸载后，并清除缓存数据，再次安装，此时isNew = 1，不准确
 * 2). 手机存储不足，清除app缓存，再次访问时isNew = 1，不准确
 * 3). 只有缓存存在时，判断用户不是新访客时，isNew = 0，准确的
 */
public class AdjustIsNewProcessFuncation extends KeyedProcessFunction<String, String, String> {
    // step1、声明状态
    private transient ValueState<String> firstDateState = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        // step2、初始化状态
        firstDateState = getRuntimeContext().getState(new ValueStateDescriptor<String>("first-date", String.class));
    }

    @Override
    public void processElement(String value, Context ctx, Collector<String> out) throws Exception {
        //step3、获取状态值
        String firstDate = firstDateState.value();
        // 解析字段值
        JSONObject jsonObject = JSON.parseObject(value);
        String isNewValue = jsonObject.getJSONObject("common").getString("is_new");
        Long tsValue = jsonObject.getLong("ts");
        // isNew = 1 --->  firstDate = null, firstDate = 有值
        if ("1".equals(isNewValue)) {
            if (null == firstDate) {
                //step4、更新状态值
                // 此时is_new正确，状态值更新为第一次访问日期
                firstDateState.update(DateTimeUtil.convertLongToString(tsValue, DateTimeUtil.DATE_FORMAT));
            } else {
                // 状态值（首次访问日期）等于数据日期，不做处理
                if (!firstDate.equals(DateTimeUtil.convertLongToString(tsValue, DateTimeUtil.DATE_FORMAT))) {
                    // 此时is_new属于异常，需要纠正，设置为0
                    jsonObject.getJSONObject("common").put("is_new", "0");
                }
            }
        }
        // isNew = 0 --->  firstDate = null, firstDate = 有值
        else {
            if (null == firstDate) {
                firstDateState.update(DateTimeUtil.convertLongToString(tsValue - 86400000, DateTimeUtil.DATE_FORMAT));
            }
        }
        //输出数据
        out.collect(jsonObject.toJSONString());
    }
}
